package com.ichoice.rabbitmq.rpc.helper.service;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ClassUtil;
import cn.hutool.core.util.ReflectUtil;
import com.ichoice.rabbitmq.rpc.helper.consts.CommonContant;
import com.ichoice.rabbitmq.rpc.helper.model.CallBackModel;
import com.ichoice.rabbitmq.rpc.helper.model.DelayMessagePoster;
import com.ichoice.rabbitmq.rpc.helper.model.ProducerModel;
import com.ichoice.rabbitmq.rpc.helper.util.MqMetadataUtils;
import com.ichoice.rabbitmq.rpc.helper.util.SpringUtil;
import com.ichoice.rabbitmq.rpc.helper.util.StringUit;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Objects;

/**
 * 服务端消费实现
 *
 * @author xianrensawang
 * @since 2021/1/28
 */
@Slf4j
public class Consumer {

    private final RabbitTemplate rabbitTemplate;

    public Consumer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void consumer(ProducerModel producerModel) throws Exception {
        String beanName = StringUit.firstCharLower(producerModel.getBean());
        String method = producerModel.getMethod();
        CallBackModel result;
        try {
            Object bean = SpringUtil.getBean(beanName);
            Method[] methods = ReflectUtil.getMethods(bean.getClass());
            Object invoke = "";
            for (Method m : methods) {
                if (Objects.equals(m.getName(), method)) {
                    Class<?>[] parameterTypes = m.getParameterTypes();
                    if (parameterTypes.length != producerModel.getArgs().length) {
                        continue;
                    }
                    Object[] args = new Object[parameterTypes.length];
                    Object[] params = producerModel.getArgs();
                    for (int i = 0; i < parameterTypes.length; i++) {
                        boolean primitiveWrapper = ClassUtil.isPrimitiveWrapper(parameterTypes[i]);
                        boolean basicType = ClassUtil.isBasicType(parameterTypes[i]);
                        boolean isCollection = Collection.class.isAssignableFrom(parameterTypes[i]);
                        boolean isString = String.class.isAssignableFrom(parameterTypes[i]);
                        if (!primitiveWrapper && !basicType && !isString && !isCollection) {
                            Object o = BeanUtil.toBeanIgnoreError(params[i], parameterTypes[i]);
                            args[i] = o;
                        } else {
                            // 如果方法入参是long型,进行数据转换,这里是json序列化的一个问题，Long型重新序列化会变成Integer
                            if (Long.class.isAssignableFrom(parameterTypes[i])) {
                                args[i] = Long.valueOf(params[i].toString());
                            } else {
                                args[i] = params[i];
                            }
                        }
                    }
                    invoke = m.invoke(bean, args);
                    break;
                }
            }
            if (invoke == "") {
                invoke = "执行入参失败,可能是指定服务bean对应方法跟入参不匹配,bean:" + producerModel.getBean() + ";method:" + producerModel.getMethod();
            }
            result = CallBackModel.success(producerModel.getAck(), invoke);
            back(producerModel, result);
            log.info("消息队列rpc调用成功,调用端服务名:{},服务端bean:{},服务端方法:{},",producerModel.getCurrentApplication(),producerModel.getTargetApplication(),
                    producerModel.getMethod());
        } catch (Exception e) {
            result = CallBackModel.fail(producerModel.getAck(), e.getMessage());
            log.error("消息队列rpc调用失败,ackId:{},调用端:{},服务端:{},异常信息:{}", producerModel.getAck(), producerModel.getCurrentApplication(),
                    producerModel.getTargetApplication(), StringUit.toStringNewLine(e.getStackTrace()));
            if (Objects.isNull(producerModel.getRetry())) {
                log.error("该入参未配置重试,消息ackId：{}", producerModel.getAck());
                back(producerModel, result);
                throw e;
            } else if (Objects.equals(producerModel.getRetry(), 0)) {
                log.error("该入参重试次数到达上限,消息ackId:{},消息重试次数:{},消息重试延迟:{}", producerModel.getAck(), producerModel.getHasRetry(), producerModel.getRetryDelay());
                back(producerModel, result);
                throw e;
            } else {
                Integer retry = producerModel.getRetry();
                retry--;
                producerModel.setRetry(retry);
                Integer hasRetry = Objects.isNull(producerModel.getHasRetry()) ? 0 : producerModel.getHasRetry();
                hasRetry++;
                producerModel.setHasRetry(hasRetry);
                log.error("即将开始第{}次重试,消息ackId:{},消息重试次数:{},消息重试延迟:{}", hasRetry, producerModel.getAck(), producerModel.getRetry(), producerModel.getRetryDelay());
                retrySent(producerModel);
            }
        }
    }

    private void back(ProducerModel producerModel, CallBackModel result) {
        if (producerModel.getNeedRes()) {
            result.setCallBackBean(producerModel.getCallBackBean());
            result.setCallBackMethod(producerModel.getCallBackMethod());
            result.setCurrentApplication(producerModel.getCurrentApplication());
            result.setTargetApplication(producerModel.getTargetApplication());
            result.setAck(producerModel.getAck());
            rabbitTemplate.convertAndSend(producerModel.getCallBackExchange(), producerModel.getCallBackRoutKey(), result);
        }
    }


    /**
     * 重试发送,到延迟队列
     *
     * @param producerModel 发送入参
     */
    private void retrySent(ProducerModel producerModel) {
        String target = producerModel.getTargetApplication();
        String exchange = CommonContant.EXCHANGE_DELAY;
        String key = MqMetadataUtils.selfDelayKey(target);
        String retryDelay = producerModel.getRetryDelay();
        if (Strings.isBlank(retryDelay)) {
            // 默认 100ms
            retryDelay = "100";
            producerModel.setRetryDelay(retryDelay);
        }
        rabbitTemplate.convertAndSend(exchange, key, producerModel, new DelayMessagePoster(Long.valueOf(retryDelay)));
    }

}
